package com.renrendoc.network.flow.sink;

import com.google.gson.Gson;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.lang.reflect.Field;
import java.sql.Array;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import static com.renrendoc.network.flow.util.SocketClient.callRemoteMethod;
import static com.renrendoc.network.flow.util.SocketClient.createRequestParams;

public class RpcSink<T> extends RichSinkFunction<T> {
    private  String cls;
    private  String method;

    public RpcSink(String cls, String method) {
        this.cls = cls;
        this.method = method;
    }

    @Override
    public void invoke(T value, Context context) throws Exception {
//        super.invoke(value, context);
        // 创建 Gson 对象
        Gson gson = new Gson();
        String jsonString = gson.toJson(value);

        try {
            // 调用远程方法
            Map<String, Object> responseData = callRemoteMethod(createRequestParams(this.cls, this.method, new Object[]{jsonString}, "flink"));
//            System.out.println("Response data from server: " + responseData.get("aid"));
        } catch (Exception e) {
            System.out.println("Exception occurred: " + e.getMessage());
        }

    }

    public static Object[] objectArrayFromEvent(Object obj) {
        List<Object> list = new ArrayList<>();
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getFields(); // 获取所有公共属性
        for (Field field : fields) {
            try {
                list.add(field.get(obj)); // 将属性值添加到列表中
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        }
        return list.toArray(new Object[0]); // 转换为对象数组并返回
    }
}
